package com.atguigu.day05;

import com.atguigu.utils.ClickEvent;
import com.atguigu.utils.ClickSource;
import com.atguigu.utils.UrlViewCountPerWindow;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

// 使用KeyedProcessFunction实现窗口
// 实现增量聚合函数和全窗口聚合函数结合使用的情况
public class Example3 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickSource())
                .keyBy(r -> r.url)
                // 滚动窗口的大小是5000毫秒
                .process(new FakeWindow(5000L))
                .print();

        env.execute();
    }

    public static class FakeWindow extends KeyedProcessFunction<String, ClickEvent, UrlViewCountPerWindow> {
        // 窗口的长度
        private Long windowSize;

        public FakeWindow(Long windowSize) {
            this.windowSize = windowSize;
        }

        // KEY：窗口的开始时间
        // VALUE：统计窗口中的url浏览次数的累加器
        // mapState是每个key(url)独有的
        // mapState保存的是每个url对应的所有窗口信息
        private MapState<Long, Long> mapState;
        @Override
        public void open(Configuration parameters) throws Exception {
            super.open(parameters);
            mapState = getRuntimeContext().getMapState(
                    new MapStateDescriptor<Long, Long>(
                            "map-state",
                            Types.LONG,
                            Types.LONG
                    )
            );
        }

        @Override
        public void processElement(ClickEvent value, Context ctx, Collector<UrlViewCountPerWindow> out) throws Exception {
            // 根据数据的到达机器的时间，可以唯一确定数据属于哪一个窗口
            // 窗口开始时间 = 时间戳 - 时间戳 % 窗口长度
            // 20 = 21 - 21 % 5
            // 20 = 22 - 22 % 5
            // 所以时间戳是21秒的数据属于（20～25）
            long currTs = ctx.timerService().currentProcessingTime();
            long windowStart = currTs - currTs % windowSize;
            long windowEnd = windowStart + windowSize;
            if (!mapState.contains(windowStart)) {
                // 当mapState不包含windowStart这个key
                // 说明windowStart对应的窗口的第一条数据到达了
                mapState.put(windowStart, 1L);
            } else {
                // windowStart这个key已经在mapState中了
                // 那么，来一条数据，累加器加一
                // 聚合规则
                mapState.put(windowStart, mapState.get(windowStart) + 1L);
            }

            // 怎么将窗口闭合计算？
            // 在窗口结束时间-1毫秒注册一个定时器
            // 窗口是左闭右开区间
            // 只有第一次注册起作用！！！
            ctx.timerService().registerProcessingTimeTimer(windowEnd - 1L);
        }

        @Override
        public void onTimer(long timestamp, OnTimerContext ctx, Collector<UrlViewCountPerWindow> out) throws Exception {
            super.onTimer(timestamp, ctx, out);
            // 定时器用来计算窗口，并删除窗口
            // timestamp是定时器触发的时间戳，也就是windowEnd - 1毫秒
            long windowEnd = timestamp + 1L;
            long windowStart = windowEnd - windowSize;
            // 对应了getResult函数
            long count = mapState.get(windowStart); // 获取统计值
            String url = ctx.getCurrentKey();
            mapState.remove(windowStart); // 删除窗口
            out.collect(
                    new UrlViewCountPerWindow(
                            url,
                            count,
                            windowStart,
                            windowEnd
                    )
            );
        }
    }
}
